package com.bungarus;

import com.bungarus.busi.InnerClient;
import com.bungarus.busi.TransferInnerClient;
import com.bungarus.busi.TransferRemedy;
import com.bungarus.busi.TransferServerChildHandlerInitializer;
import com.bungarus.cluster.ClusterInfo;
import com.bungarus.cluster.NodeInfo;
import com.bungarus.model.AbstractBungarusServer;
import com.bungarus.model.AbstractBungarusServerWithMultipleClients;
import com.bungarus.model.BungarusClientException;
import com.bungarus.model.BungarusServerException;
import io.netty.channel.ChannelHandler;
import org.apache.log4j.Logger;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.List;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledThreadPoolExecutor;

/**
 * Created by tang on 2018/11/24.
 */
public class Transfer extends AbstractBungarusServerWithMultipleClients {
    Logger logger = Logger.getLogger(Transfer.class);
    private static final long SAFEGUARD_INIT_DELAY = 60 * 1000;
    private static final long SAFEGUARD_RUNNING_INTERVAL = 30 * 1000;

    private static final int PORT_7600 = 7600;

    public static final String TAG = "transfer";

    /**
     * clusters connected to as client
     */
    private static final String[] CLUSTER_TAGS = {CLUSTER_TAG_BUSI};

    private ScheduledExecutorService safeGuard = new ScheduledThreadPoolExecutor(1);

    public static void main(String[] args) throws Exception {
        int[] messagePorts = {PORT_7600};

        Transfer transfer = new Transfer();
        transfer.bind(System.getProperty(AbstractBungarusServer.HOST), messagePorts, null);
        transfer.initClusterAndConnectRemoteNode();
        transfer.initRemedy();
        transfer.listen(null);
    }

    @Override
    public InnerClient connect(int index, CopyOnWriteArrayList<NodeInfo> queue) throws BungarusClientException {
        InnerClient client = new TransferInnerClient(this, index, queue, getMongoTemplate4FailedRetry());
        client.run(getTag());
        return client;
    }

    @Override
    protected InnerClient createNewInnerClientOnClusterNodeUpdated(NodeInfo node) {
        return new TransferInnerClient(this, 0, new CopyOnWriteArrayList<>(Collections.singleton(node)), this.getMongoTemplate4FailedRetry());
    }

    @Override
    public void initRemedy() throws Exception {
        TransferRemedy.start(this);
//        safeGuard.scheduleAtFixedRate(() -> {
//            try {
//                List<ClusterInfo> clusters = getClusterManager().clusters(CLUSTER_TAG_BUSI);
//                //Supports only one cluster.
//                //TODO: supports multiple cluster
//                if(!clusters.isEmpty()) {
//                    List<NodeInfo> latestNodesInCluster = getClusterManager().nodes(CLUSTER_TAG_BUSI, clusters.get(0).getClusterID());
//                    logger.debug("Safeguard is repairing the connections to busi nodes => size = " + latestNodesInCluster.size());
//                    refreshConnectedClients(CLUSTER_TAG_BUSI, latestNodesInCluster);
//                }
//            }
//            catch (Exception e) {
//                e.printStackTrace();
//            }
//        }, SAFEGUARD_INIT_DELAY, SAFEGUARD_RUNNING_INTERVAL, TimeUnit.MILLISECONDS);
    }

    @Override
    public ChannelHandler getChildHandlerInitializer() {
        return new TransferServerChildHandlerInitializer(this);
    }

    @Override
    public String getTag() {
        return TAG;
    }

    @Override
    public String tagPortString4Connection() {
        return "(access|7600)";
    }

    @Override
    public String[] getRemoteNodeTags() {
        return Arrays.copyOf(CLUSTER_TAGS, CLUSTER_TAGS.length);
    }

    @Override
    public CopyOnWriteArrayList<NodeInfo> calcRemoteServerNodes(CopyOnWriteArrayList<NodeInfo> originList) {
        return originList;
    }

    @Override
    public void shutdown() throws BungarusServerException {
        if(!safeGuard.isShutdown()) {
            safeGuard.shutdown();
        }
        super.shutdown();
    }

    /**
     * Get all remote nodes in all clusters (busi clusters).
     * Every node in transfer cluster must connect to all remote nodes in all clusters (busi clusters)
     */
    @Override
    protected void getNodesInCluster() {
        List<ClusterInfo> clusters;
        for (String CLUSTER_TAG : getRemoteNodeTags()) {
            try {
                clusters = getClusterManager().clusters(CLUSTER_TAG);
                if (null != clusters && clusters.size() > 0) {
                    List<NodeInfo> nodes = new ArrayList<>();
                    for(ClusterInfo cluster: clusters) {
                        nodes.addAll(getClusterManager().nodes(CLUSTER_TAG, cluster.getClusterID()));
                    }
                    super.remoteServerNodes.put(CLUSTER_TAG, new CopyOnWriteArrayList<>(nodes));
                }
            }
            catch (Exception e) {
                logger.info(e.getMessage());
                e.printStackTrace();
            }
        }
    }
}
